package jupitermouse.site.keytuple

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

object PojoStreamingWCSCalaApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val streamSet = env.socketTextStream("localhost", 9999)

    import org.apache.flink.api.scala._
    streamSet
      .flatMap(_.toLowerCase.split(" "))
      .filter(_.nonEmpty)
      .map(x => WC(x,1))
//      .keyBy("word")
      .keyBy(_.word)
      .timeWindow(Time.seconds(5))
      .sum("count")
      .print()
      .setParallelism(1)

    env.execute();

  }

}
case class WC(var word: String, var count: Int)
